package com.yunhe.transfer.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.model.authority.Station;
import com.yunhe.transfer.client.AuthorityClient;
import com.yunhe.transfer.client.DeviceClient;
import com.yunhe.transfer.client.MeasHistoryClient;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.model.DomainList;
import com.yunhe.common.model.data.Measurement;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * todo 由于站端程序未发布，临时处理上海电站实时数据保存到历史量测，站端发布以后需要去掉
 */
@Component
public class MeasurementKafkaHandler {

    @Autowired
    private MeasHistoryClient measHistoryClient;
    @Autowired
    private AuthorityClient authorityClient;
    @Autowired
    private DeviceClient deviceClient;

    private Logger logger = LoggerFactory.getLogger(MeasurementKafkaHandler.class);

    private static final ObjectMapper mapper = new ObjectMapper();

    @KafkaListener(topics = KafkaConsts.Topic.Realtime + "Shanghai")
    public void consumer(ConsumerRecord<String, String> record) {
        long rtime = System.currentTimeMillis();
        Station station = authorityClient.getStationByCode(1000L).getBody().getResults();
        if (station == null || (station.getUseHeartbeat() != null && !station.getUseHeartbeat())) {
            return;
        }
        try {
            String value = record.value();
            Map map = mapper.readValue(value, Map.class);
            if (map != null && map.size() > 0) {
                List<Measurement> measurements = new ArrayList<>();
                map.forEach((pid, val) -> {
                    try {
                        long pointNumber = Long.parseLong(pid.toString());
                        Map<Long, Object> ignorePids = getIgnorePids();
                        if (!ignorePids.containsKey(pointNumber)) {
                            Measurement measurement = new Measurement();
                            measurement.setPointNumber(pointNumber);
                            measurement.setVal(Double.parseDouble(val.toString()));
                            measurement.setRtime(rtime);
                            measurements.add(measurement);
                        }
                    } catch (Exception e) {
                        logger.error("量测转换失败{},{}", pid, val);
                    }
                });
                if (measurements.size() > 0) {
                    measHistoryClient.save(new DomainList<>(measurements), MeasurementConsts.FREQUENCY.Original.value());
                }
            }
        } catch (Exception e) {
            logger.error("MeasurementKafkaHandler处理数据失败", e);
        }
    }

    private Map<Long, Object> getIgnorePids() {
        Map<Long, Object> result = new HashMap<>();
        try {
            List<Long> ignoreAnalogs = deviceClient.getNeedStorageAnalogs(false).getResults();
            if (ignoreAnalogs != null && ignoreAnalogs.size() > 0) {
                for (Long pointNumber : ignoreAnalogs) {
                    result.put(pointNumber, null);
                }
            }
        } catch (Exception e) {
            logger.error("获取不需要切数据点号出错", e);
        }
        return result;
    }
}
